package com.flink.wc.demo.demo_20240801;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class Demo_05 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.BATCH); //设置运行模式-批处理模式
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        DataStreamSource<Row> source = env.fromElements(
                Row.of("张三", 13),
                Row.of("李四", 14),
                Row.of("王五", 15),
                Row.of("王五", 20),
                Row.of("赵六", 16),
                Row.of("田七", 17),
                Row.of("田七", 17)
        );
        Table nameTable = tableEnv.fromDataStream(source).as("name", "age");
        tableEnv.createTemporaryView("nameTable",nameTable);
        Table ageSumTable = tableEnv.sqlQuery("select name,sum(age) from nameTable group by name");
        tableEnv.toDataStream(ageSumTable).writeAsText("E:\\mywork\\FlinkDemo-1.19\\src\\main\\java\\com\\flink\\wc\\demo\\demo_20240801\\test_out2.txt");
        env.execute();
    }
}
